/* Handles request for communication through APR sockets-explain*/
#include <stdlib.h>
#include <stdio.h>
#include <apr_general.h>
#include <apr_pools.h>
#include <apr_network_io.h>
#include <apr_poll.h>
#include <apr_thread_proc.h>
#include <apr_thread_cond.h>
#include <apr_thread_mutex.h>
#include "mpi.h"
#include "oom.h"
#include "cm.h"
#include "msg.h"
#include "env.h"
#include "log.h"
#include "util.h"
#include "dtypes.h"

#define CM_BACKLOG		5
#define CM_MAX_REQUESTS	100

static apr_pool_t *cm_pool = NULL;
static apr_socket_t *listen_socket = NULL;
static apr_socket_t **send_sockets = NULL;
static apr_pollfd_t **poll_fds = NULL;
static apr_pollset_t *pollset = NULL;
static apr_thread_t *mgr_thread = NULL;
static int mgr_thread_should_run = 0;

static apr_thread_mutex_t *request_queue_mutex;
static apr_thread_cond_t *request_queue_cond;
static Request *request_queue[CM_MAX_REQUESTS];
static int request_queue_len = 0;

typedef struct {
	int source_world_rank;
	Envelope envelope;
	int message_size;
} SendReq;

static apr_thread_mutex_t *sendreq_queue_mutex;
static apr_thread_cond_t *sendreq_queue_cond;
static SendReq *sendreq_queue[CM_MAX_REQUESTS];
static int sendreq_queue_len = 0;

static void * APR_THREAD_FUNC handle_requests(apr_thread_t *thread, void *data);
static void add_pollfd(apr_socket_t *socket, int rank);
static apr_socket_t *get_connection(int world_rank);
static void open_connection(int dest_world_rank);
static void close_connection(int world_rank);
static void accept_connection();
static void handle_message(apr_socket_t *socket, int source_world_rank);
static void handle_request(Request *request);
static void enqueue_request(Request *request);
static void dequeue_request(Request *request);
static void enqueue_sendreq(int source_world_rank, Envelope *envelope,
	int message_size);
static void dequeue_sendreq(SendReq *sendreq);
static Request *find_request(RequestType type, RequestStatus status,
	Envelope *request_env, Envelope *matched_env);
static SendReq *find_sendreq(Envelope *request_env);

int CM_Init(int *listen_port)
{
	apr_status_t status;
	apr_sockaddr_t *sa;
	int world_size = get_world_size();
	apr_pollfd_t *new_pollfd;
	
	/* allocate a memory pool */
	
	status = apr_pool_create(&cm_pool, NULL);
	if (status != APR_SUCCESS) return CM_FAILURE;
	
	/* allocate an array to hold sockets connecting us to other processors */
	
	send_sockets = (apr_socket_t **) apr_palloc(cm_pool, sizeof(apr_socket_t *)
		* world_size);
	
	/* create our listening socket and start listening for incoming
	   connections */
		
	status = apr_socket_create(&listen_socket, APR_INET, SOCK_STREAM,
		APR_PROTO_TCP, cm_pool);
	if (status != APR_SUCCESS) return CM_FAILURE;

	status = apr_sockaddr_info_get(&sa, APR_ANYADDR, APR_UNSPEC, 0, 0, cm_pool);
	if (status != APR_SUCCESS) return CM_FAILURE;

	status = apr_socket_bind(listen_socket, sa);
	if (status != APR_SUCCESS) return CM_FAILURE;
	
	status = apr_socket_listen(listen_socket, CM_BACKLOG);
	if (status != APR_SUCCESS) return CM_FAILURE;
	
	/* find out what port our listening socket is bound to and return the
	   port number */
	
	status = apr_socket_addr_get(&sa, APR_LOCAL, listen_socket);
	if (status != APR_SUCCESS) return CM_FAILURE;
	
	*listen_port = sa->port;
	
	/* allocate an array to hold polling descriptors for all active sockets;
	   the descriptor for the socket connecting to host X is at index X and
	   the descriptor for the listening socket goes at the end */
	
	poll_fds = (apr_pollfd_t **) apr_palloc(cm_pool, sizeof(apr_pollfd_t *)
		* (world_size + 1));
	
	/* allocate a poll set */
	
	status = apr_pollset_create(&pollset, world_size + 1, cm_pool, 0);
	if (status != APR_SUCCESS) return CM_FAILURE;
	
	/* create the poll descriptor for the listening socket and add it to the
	   pollset */
	
	new_pollfd = (apr_pollfd_t *) apr_palloc(cm_pool, sizeof(apr_pollfd_t));
	new_pollfd->p = cm_pool;
	new_pollfd->desc_type = APR_POLL_SOCKET;
	new_pollfd->reqevents = APR_POLLIN;
	new_pollfd->rtnevents = 0;
	new_pollfd->desc.s = listen_socket;
	new_pollfd->client_data = NULL;

	poll_fds[world_size] = new_pollfd;
	status = apr_pollset_add(pollset, new_pollfd);
	if (status != APR_SUCCESS) return CM_FAILURE;

	/* create a mutex for access to the request queue */
	
	status = apr_thread_mutex_create(&request_queue_mutex, APR_THREAD_MUTEX_NESTED,
		cm_pool);
	status = apr_thread_mutex_create(&sendreq_queue_mutex, APR_THREAD_MUTEX_NESTED,
			cm_pool);
	if (status != APR_SUCCESS) return CM_FAILURE;
	
	/* create a condition variable for signaling the removal of an entry
	   from the queue (in the case that the queue is full) */
	
	status = apr_thread_cond_create(&request_queue_cond, cm_pool);
	status = apr_thread_cond_create(&sendreq_queue_cond, cm_pool);
	
	if (status != APR_SUCCESS) return CM_FAILURE;
		
	/* start up the manager thread */
	
	mgr_thread_should_run = 1;
	status = apr_thread_create(&mgr_thread, NULL, handle_requests, NULL, cm_pool);
	if (status != APR_SUCCESS) return CM_FAILURE;
	
	return CM_SUCCESS;
}

int CM_Finalize()
{
	apr_status_t status;
	int i;
	
	mgr_thread_should_run = 0;
	status = apr_thread_join(&status, mgr_thread);

	status = apr_pollset_destroy(pollset);
	
	for (i = 0; i < get_world_size(); i++)
	{
		if (send_sockets[i] != NULL) apr_socket_close(send_sockets[i]);
	}
	
	status = apr_socket_shutdown(listen_socket, APR_SHUTDOWN_READWRITE);
	
	apr_pool_destroy(cm_pool);

	return 0;
}

int CM_Send(void *buf, int count, MPI_Datatype datatype, int source, int dest,
	int context, int tag, int dest_world_rank, MPI_Request *handle)
{
	Request *request;

	/* allocate a new request object */
	
	*handle = OOM_request_create();
	request = OOM_request_lookup(*handle);
	if (request == NULL) return CM_FAILURE;
	
	/* set the request object fields for a send request */

	request->req_type = SEND;
	request->buf = buf;
	request->count = count;
	request->datatype = datatype;
	request->request_env.source = source;
	request->request_env.dest = dest;
	request->request_env.context = context;
	request->request_env.tag = tag;
	request->dest_world_rank = dest_world_rank;
	MPI_Type_size(datatype, &(request->num_bytes));
	request->num_bytes *= count;
	debug("send request allocated and initialized");
	
	/* add the new request object to the queue */
	
	CM_Start(request);
	debug("send request started");
	
	return CM_SUCCESS;
}

/**********************************************
 * \author: James Edmondson, Zach Lowry
 * \brief: Sends data from this process to all other process in the
 *         Communicator. The Request ** stuff was added by Zach to
 *         anticipate needing a complete list of request objects to
 *         use later. The functionality is not currently used or
 *         implemented in this function.
 **********************************************/

int CM_SendAll(void *buf, int count, MPI_Datatype datatype, int source,
        Comm* comm, Request **requests)
{
	Group *real_group;
	MPI_Request request;
	Request *real_request;
	int my_group_rank;
	int dest;
	int tag = MPI_ANY_TAG;
	int tsource = source;
	int ierr = 0;
	
	if (comm == NULL) return MPI_ERR_COMM;
	real_group = OOM_group_lookup(comm->group);
	if (real_group == NULL) return MPI_ERR_COMM;

	my_group_rank = comm->source;

	requests = (Request **) apr_palloc(cm_pool, 
			sizeof(real_request) * real_group->num_members);

	debug("SENDALL: members %d, rank %d",real_group->num_members,
			my_group_rank);
	
		for (dest = 0; dest < real_group->num_members; dest++)
		{
			debug("SENDALL: dest %d",dest);
			if( dest != my_group_rank )
			{
				debug("SENDALL: dest %d passed",dest);
				CM_Send(buf,count,datatype,source,dest,
						comm->coll_context,tag,
						real_group->members[dest],&request);
				debug("SENDALL: now we call wait %d",dest);
				real_request = OOM_request_lookup(request);
				CM_Wait(real_request,&tsource,&tag,&ierr);
				// requests[dest] = real_request;
				debug("SENDALL: we finished wait.");
			}
		}
	
	return CM_SUCCESS;

}

int CM_Recv(void *buf, int count, MPI_Datatype datatype, int source,
	int dest, int context, int tag, MPI_Request *handle)
{
	Request *request;

	/* allocate a new request object */
	
	*handle = OOM_request_create();
	request = OOM_request_lookup(*handle);
	if (request == NULL) return CM_FAILURE;
	
	/* set the request object fields for a receive request */
	
	request->req_type = RECV;
	request->buf = buf;
	request->count = count;
	request->datatype = datatype;
	request->request_env.source = source;
	request->request_env.dest = dest;
	request->request_env.context = context;
	request->request_env.tag = tag;
	debug("recv request allocated and initialized");
	
	/* enqueue the send request */
	
	CM_Start(request);
	debug("recv request started");
	
	return CM_SUCCESS;
}

/**********************************************
 * \author: Zach Lowry
 * \brief: Receives data from each process and places
 *         it into the buf. Currently only used in
 *         MPI_Scatter.
 **********************************************/

int CM_RecvAll(void *buf, int count, MPI_Datatype datatype,
	int dest, Comm *comm, Request **requests)
{
	Group *real_group;
	MPI_Request request;
	Request *real_request;
	int my_group_rank;
	int source = 0;
	int tag = 1;
	int tsource = source;
	int ierr = 0;
	int stride = sizeof_base_type(datatype) * count;
	char *recv_buf;
	
	if (comm == NULL) return MPI_ERR_COMM;
	real_group = OOM_group_lookup(comm->group);
	if (real_group == NULL) return MPI_ERR_COMM;

	my_group_rank = comm->source;


	//requests = (Request *) apr_palloc(cm_pool, sizeof(real_request) * real_group->num_members);

	
		for (source = 0; source < real_group->num_members; source++)
		{
			recv_buf = (char*)buf + (source * stride);
			if( source != my_group_rank )
			{
				CM_Recv(recv_buf,count,datatype,source,dest,
						comm->coll_context,tag,
						&request);
				real_request = OOM_request_lookup(request);
	//			requests[source] = real_request;
				CM_Wait(real_request,&tsource,&tag,&ierr);
			}
		}
	
	return CM_SUCCESS;

}

int CM_Wait(Request *request, int *source, int *tag, int *error)
{
	/* lock the request to allow use of the condition variable */
	
	apr_thread_mutex_lock(request->mutex);

	/* wait on the condition variable until the request is marked
	   as complete.  the condition variable will be signaled when
	   the request is marked as complete. */
	
	while (request->status != COMPLETE)
	{
		debug("waiting for request completion");
		apr_thread_cond_wait(request->cond, request->mutex);
	}
	debug("request completed");
	
	/* set the return paramters from the request.  these values will
	   be used to set the MPI_Status for this request */
	
	*source = request->matched_env.source;
	*tag = request->matched_env.tag;
	*error = MPI_SUCCESS;
	
	/* unlock the request mutex now that we've finished with it */
	
	apr_thread_mutex_unlock(request->mutex);
	
	return CM_SUCCESS;
}

int CM_Wait2(Request *request, MPI_Status *status)
{
	/* lock the request to allow use of the condition variable */
	
	apr_thread_mutex_lock(request->mutex);

	/* wait on the condition variable until the request is marked
	   as complete.  the condition variable will be signaled when
	   the request is marked as complete. */
	
	while (request->status != COMPLETE)
	{
		debug("waiting for request completion");
		apr_thread_cond_wait(request->cond, request->mutex);
	}
	debug("request completed");
	
	/* set the status paramters from the request */
	
	set_status_from_request(status, request);
	
	/* unlock the request mutex now that we've finished with it */
	
	apr_thread_mutex_unlock(request->mutex);
	
	return CM_SUCCESS;
}

int CM_Waitany(int count, Request **requests, int *index, MPI_Status *status)
{
	int completed, i;
	
	apr_thread_mutex_lock(request_queue_mutex);
	
	completed = 0;
	while (! completed)
	{
		for (i = 0; i < count; i++)
		{
			if (requests[i] != NULL && requests[i]->status == COMPLETE)
			{
				completed = 1;
				*index = i;
				set_status_from_request(status, requests[i]);
				break;
			}
		}
		if (! completed)
		{
			apr_thread_cond_wait(request_queue_cond, request_queue_mutex);
		}
	}
	
	apr_thread_mutex_unlock(request_queue_mutex);
	
	return CM_SUCCESS;
}

int CM_Waitsome(int incount, Request **requests, int *outcount, int *indices, MPI_Status *statuses)
{
	int i;
	
	apr_thread_mutex_lock(request_queue_mutex);

	*outcount = 0;
	while (*outcount == 0)
	{
		for (i = 0; i < incount; i++)
		{
			fprintf(stderr,"incount = %d, i = %d, and *outcount = %d.\n",
					incount,i,*outcount);
			if (requests[i] != NULL && requests[i]->status == COMPLETE)
			{
				indices[*outcount] = i;
				set_status_from_request(statuses + *outcount, requests[i]);
				(*outcount)++;
			}
		}
		if (*outcount == 0)
		{
			fprintf(stderr,"Going into cond wait.\n");
			apr_thread_cond_wait(request_queue_cond, request_queue_mutex);
			fprintf(stderr,"Coming out of cond wait.\n");
		}
	}
	
	apr_thread_mutex_unlock(request_queue_mutex);
	
	return CM_SUCCESS;
}

int CM_Test(Request *request, int *flag, int *source, int *tag, int *error)
{
	/* lock the request mutex to ensure fields aren't written
	   while we are reading them */
	
	apr_thread_mutex_lock(request->mutex);
	
	/* set the flag based on the completion status of the request */
	
	*flag = (request->status == COMPLETE);
	
	/* set the other return parameters from the request object.  these
	   values will be used to set the MPI_Status for this request */
	
	*source = request->matched_env.source;
	*tag = request->matched_env.tag;
	*error = MPI_SUCCESS;
	
	/* unlock the request mutex now that we're done with it */
	
	apr_thread_mutex_unlock(request->mutex);
	
	return CM_SUCCESS;
}

int CM_Test2(Request *request, int *flag, MPI_Status *status)
{
	/* lock the request mutex to ensure fields aren't written
	   while we are reading them */
	
	apr_thread_mutex_lock(request->mutex);
	
	/* set the flag based on the completion status of the request */
	
	*flag = (request->status == COMPLETE);
	
	/* set the other return parameters from the request object */
	
	set_status_from_request(status, request);
	
	/* unlock the request mutex now that we're done with it */
	
	apr_thread_mutex_unlock(request->mutex);
	
	return CM_SUCCESS;
}

int CM_Iprobe(int source, int dest, int context, int tag, int *flag,
	MPI_Status *status)
{
	Envelope request_env;
	SendReq *sendreq;
	
	/* initialize the request envelope from the input parameters (to be used
	   with find_sendreq) */
	
	request_env.source = source;
	request_env.dest = dest;
	request_env.context = context;
	request_env.tag = tag;

	/* assume no match */
	
	*flag = 0;
	
	/* check the queue of unmatched send requests for any that match
	   this request's envelope.  if one is found, set the output paramaters
	   from the matched envelope */

	apr_thread_mutex_lock(sendreq_queue_mutex);
	sendreq = find_sendreq(&request_env);
	if (sendreq != NULL)
	{
		debug("found a matching sendreq for probe");
		*flag = 1;
		if (status != MPI_STATUS_IGNORE)
		{
			status->MPI_SOURCE = sendreq->envelope.source;
		 	status->MPI_TAG = sendreq->envelope.tag;
			status->bytes_recvd = sendreq->message_size;
			status->MPI_ERROR = MPI_SUCCESS;
		}
	}
	apr_thread_mutex_unlock(sendreq_queue_mutex);
	
	return CM_SUCCESS;
}

int CM_Probe(int source, int dest, int context, int tag, MPI_Status *status)
{
	int flag;
	
	/* no match at the start */
	
	flag = 0;
	
	/* do a nonblocking probe for the given parameters */
	
	CM_Iprobe(source, dest, context, tag, &flag, status);
	
	/* if the initial probe returned no match, wait until the sendreq queue
	   is changed and try again (and keep doing so until we get a match) */
		
	while (!flag)
	{
		debug("waiting for sendreq queue to change");
		apr_thread_mutex_lock(sendreq_queue_mutex);
		apr_thread_cond_wait(sendreq_queue_cond, sendreq_queue_mutex);
		apr_thread_mutex_unlock(sendreq_queue_mutex);
		debug("sendreq queue change signaled");
		
		CM_Iprobe(source, dest, context, tag, &flag, status);
	}
	
	return CM_SUCCESS;
}

int CM_Start(Request *request)
{
	SendReq *sendreq;
	
	if (request->req_type == SEND)
	{
		request->status = NEW;
	}
	else if (request->req_type == RECV)
	{
		/* check the queue of unmatched send requests for any that match
		   this request's envelope.  if one is found, mark this request
		   as matched and remove the matching send request from the queue */

		apr_thread_mutex_lock(sendreq_queue_mutex);
		sendreq = find_sendreq(&(request->request_env));
		if (sendreq == NULL)
		{
			request->status = NEW;
		}
		else
		{
			debug("found sendreq matching new request; marking as matched");
			request->status = PENDING;
			request->source_world_rank = sendreq->source_world_rank;
			msg_env_copy(&(request->matched_env), &(sendreq->envelope));
			dequeue_sendreq(sendreq);
		}
		apr_thread_mutex_unlock(sendreq_queue_mutex);
	}
		
	/* add the new request object to the queue */
	
	enqueue_request(request);
	
	return CM_SUCCESS;
}

static void * APR_THREAD_FUNC handle_requests(apr_thread_t *thread, void *data)
{
	apr_int32_t num_signals = 0;
	const apr_pollfd_t *signaled_desc;
	int i;
	
	while ( mgr_thread_should_run )
	{
		apr_pollset_poll(pollset, .05 * APR_USEC_PER_SEC,
			   &num_signals, &signaled_desc);

		for (i = 0; i < num_signals; i++)
		{
			if (signaled_desc[i].rtnevents & APR_POLLIN)
			{
				if (signaled_desc[i].desc.s == listen_socket)
				{
					accept_connection();
				}
				else
				{
					handle_message(signaled_desc[i].desc.s,
						*((int *) signaled_desc[i].client_data));
				}
			}
		}

		apr_thread_mutex_lock(request_queue_mutex);
		for (i = 0; i < request_queue_len; i++)
		{
			handle_request(request_queue[i]);
		}
		apr_thread_mutex_unlock(request_queue_mutex);
	}

	apr_thread_exit(thread, APR_SUCCESS);
	return NULL;
}

static void add_pollfd(apr_socket_t *socket, int rank)
{
	apr_pollfd_t *new_pollfd;
	
	new_pollfd = (apr_pollfd_t *) apr_palloc(cm_pool, sizeof(apr_pollfd_t));
	
	new_pollfd->p = cm_pool;
	new_pollfd->desc_type = APR_POLL_SOCKET;
	new_pollfd->reqevents = APR_POLLIN;
	new_pollfd->rtnevents = 0;
	new_pollfd->desc.s = socket;
	new_pollfd->client_data = (int *) apr_palloc(cm_pool, sizeof(int));
	*((int *) new_pollfd->client_data) = rank;

	poll_fds[rank] = new_pollfd;
	apr_pollset_add(pollset, new_pollfd);
}

static apr_socket_t *get_connection(int target_world_rank)
{
	if (send_sockets[target_world_rank] == NULL)
	{
		open_connection(target_world_rank);
	}
	debug("got existing connection to world rank %d", target_world_rank);
	return send_sockets[target_world_rank];
}

static void open_connection(int dest)
{
	apr_status_t status;
 	apr_sockaddr_t *sa;
	apr_socket_t *s = NULL;
	int world_rank;
	apr_size_t len = sizeof(int);
	
	status = apr_sockaddr_info_get(&sa, get_processor_host(dest), APR_UNSPEC,
		get_processor_port(dest), 0, cm_pool);
	if (status != APR_SUCCESS) printf("unable to get addr info\n");
	status = apr_socket_create(&s, APR_INET, SOCK_STREAM, APR_PROTO_TCP, cm_pool);
	if (status != APR_SUCCESS) printf("unable to create socket\n");
	status = apr_socket_opt_set(s, APR_TCP_NODELAY, 1);
	apr_socket_opt_set(s, APR_SO_NONBLOCK, 0);
	apr_socket_timeout_set(s, -1);
	if (status != APR_SUCCESS) printf("unable to set options\n");
	status = apr_socket_connect(s, sa);
	if (status != APR_SUCCESS) printf("unable to connect\n");
	
	if (s == NULL)
	{
		send_sockets[dest] = NULL;
		return;
	}
	
	world_rank = get_world_rank();
	status = apr_socket_send(s, (char *) &world_rank, &len);
	if (len != sizeof(int)) printf("error writing rank\n");
	
 	send_sockets[dest] = s;
	add_pollfd(s, dest);
	
	debug("new connection opened to world rank %d", dest);
}

static void close_connection(int target_world_rank)
{
	int i;
	
	/* close the socket */
	
	apr_socket_close(send_sockets[target_world_rank]);
	
	/* remove its corresponding entry in the list of pollfds */
	
	apr_pollset_remove(pollset, poll_fds[target_world_rank]);
	poll_fds[target_world_rank] = NULL;
	
	/* remove it from our table of open sockets */
	
	send_sockets[target_world_rank] = NULL;
	
	debug("connection to world rank %d closed", target_world_rank);
}

static void accept_connection()
{
	apr_status_t status;
	apr_socket_t *new_socket;
	int *source_world_rank;
	apr_size_t len = sizeof(int);
	
	status = apr_socket_accept(&new_socket, listen_socket, cm_pool);
	if (status != APR_SUCCESS) printf("unable to accept\n");
	status = apr_socket_opt_set(new_socket, APR_TCP_NODELAY, 1);
	apr_socket_opt_set(new_socket, APR_SO_NONBLOCK, 0);
	apr_socket_timeout_set(new_socket, -1);

	source_world_rank = (int *) apr_palloc(cm_pool, sizeof(int));
		
	status = apr_socket_recv(new_socket, (char *) source_world_rank, &len);
	if (len != sizeof(int)) printf("error reading rank\n");
	
	send_sockets[*source_world_rank] = new_socket;
	add_pollfd(new_socket, *source_world_rank);
	
	debug("new connection accepted from world rank %d", *source_world_rank);
}

static void handle_message(apr_socket_t *socket, int source)
{
	apr_status_t status;
	MessageType msg_type;
	Envelope msg_env;
	int msg_size;
	Request *request;
	
	/* read the message type from the socket; if we get a status
	   of EOF, then the remote end of the socket has closed and
	   we should discard that socket */
	
	status = msg_recv_type(socket, &msg_type);
	if (APR_STATUS_IS_EOF(status))
	{
		close_connection(source);
		return;
	}
	if (status != APR_SUCCESS)
	{
		error("error reading message type");
		return;
	}
	
	/* read the envelope for this message from the socket */
	
	status = msg_recv_env(socket, &msg_env);
	if (status != APR_SUCCESS)
	{
		error("error reading message envelope");
		return;
	}
	
	/* we will need to search the queue for matching requests, so
	   we will need to lock the queue now */
	
	apr_thread_mutex_lock(request_queue_mutex);
	
	if (msg_type == SENDREQ)
	{
		debug("handling sendreq message");
		/* read the message size from the socket */

		status = msg_recv_size(socket, &msg_size);
		if (status != APR_SUCCESS)
		{
			error("error reading message size");
			return;
		}

		/* look for RECV requests that match this send request */

		request = find_request(RECV, NEW, &msg_env, NULL);
		if (request != NULL)
		{
			debug("found a matching recv request for sendreq");
			apr_thread_mutex_lock(request->mutex);
			request->source_world_rank = source;
			msg_env_copy(&(request->matched_env), &msg_env);
			request->status = PENDING;
			apr_thread_mutex_unlock(request->mutex);
			debug("recv request match recorded; request marked as pending");
		}
		else
		{
			debug("no matching recv request for sendreq found; enqueuing");
			enqueue_sendreq(source, &msg_env, msg_size);
		}
	}
	else if (msg_type == SENDOK)
	{
		debug("handling sendok message");
		/* find the recv request matching the message envelope and mark it
		   as ready */
		
		request = find_request(SEND, PENDING, &msg_env, NULL);
		if (request != NULL)
		{
			apr_thread_mutex_lock(request->mutex);
			request->status = READY;
			apr_thread_mutex_unlock(request->mutex);
			debug("send request marked as ready");
		}
		else
		{
			error("received sendok message with no matching send request");
		}
	}
	else if (msg_type == SENDMSG)
	{
		debug("handling sendmsg message");
		/* find the recv request matching the message envelope, receive the
		   message payload and mark the request as complete */
		
		request = find_request(RECV, READY, NULL, &msg_env);
		if (request != NULL)
		{
			status = msg_recv_data(socket, request->buf, request->count,
				request->datatype, &(request->num_bytes));
			if (status != APR_SUCCESS) error("error receiving message payload");
			debug("message payload received");
			apr_thread_mutex_lock(request->mutex);
			request->status = COMPLETE;
			apr_thread_cond_signal(request->cond);
			apr_thread_mutex_unlock(request->mutex);
			debug("recv request marked as complete");
			dequeue_request(request);
		}
		else
		{
			error("received sendmsg message with no matching recv request");
			/* FIXME: if we get here, there may still be the payload of the
			   bogus message in the socket.  we probably need to empty it
			   out.  OR we could just call this error fatal and bomb */
		}
	}
	else
	{
		error("unknown message type: %d", msg_type);
	}

	/* unlock the queue */
	
	apr_thread_mutex_unlock(request_queue_mutex);
}

static void handle_request(Request *request)
{
	apr_socket_t *socket;
	apr_status_t status;

	apr_thread_mutex_lock(request->mutex);
	
	if (request->req_type == SEND)
	{
		if (request->status == NEW)
		{
			debug("handling new send request");
			socket = get_connection(request->dest_world_rank);
			if (socket == NULL)
			{
				error("unable to connect to %d", request->dest_world_rank);
			}
			else
			{
				status = msg_send_type(socket, SENDREQ);
				if (status != APR_SUCCESS) error("failed to send message type");
				status = msg_send_env(socket, &(request->request_env));
				if (status != APR_SUCCESS) error("failed to send message envelope");
				status = msg_send_size(socket, request->num_bytes);
				if (status != APR_SUCCESS) error("failed to send message size");
				request->status = PENDING;
				debug("sendreq message sent; send request marked as pending");
			}
		}
		else if (request->status == READY)
		{
			debug("handling ready send request");
			socket = get_connection(request->dest_world_rank);
			if (socket == NULL)
			{
				error("unable to connect to %d", request->dest_world_rank);
			}
			else
			{
				status = msg_send_type(socket, SENDMSG);
				if (status != APR_SUCCESS) error("failed to send message type");
				status = msg_send_env(socket, &(request->request_env));
				if (status != APR_SUCCESS) error("failed to send message envelope");
				status = msg_send_data(socket, request->buf, request->count,
					request->datatype);
				if (status != APR_SUCCESS) error("failed to send message payload");
				request->status = COMPLETE;
				apr_thread_cond_signal(request->cond);
				debug("sendmsg message sent; send request marked as complete");
				dequeue_request(request);
			}
		}
	}
	else if (request->req_type == RECV)
	{
		if (request->status == PENDING)
		{
			debug("handling pending recv request");
			socket = get_connection(request->source_world_rank);
			if (socket == NULL)
			{
				error("unable to connect to %d", request->dest_world_rank);
			}
			else
			{
				status = msg_send_type(socket, SENDOK);
				if (status != APR_SUCCESS) error("failed to send message type");
				status = msg_send_env(socket, &(request->matched_env));
				if (status != APR_SUCCESS) error("failed to send message envelope");
				request->status = READY;
				debug("sendok message sent; recv request marked as ready");
			}
		}
	}
	
	apr_thread_mutex_unlock(request->mutex);
}

static void enqueue_request(Request *request)
{
	apr_thread_mutex_lock(request_queue_mutex);
	while (request_queue_len == CM_MAX_REQUESTS)
	{
		apr_thread_cond_wait(request_queue_cond, request_queue_mutex);
	}
	request_queue[request_queue_len] = request;
	request_queue_len++;
	apr_thread_cond_signal(request_queue_cond);
	apr_thread_mutex_unlock(request_queue_mutex);
}

static void dequeue_request(Request *request)
{
	int i;
	apr_thread_mutex_lock(request_queue_mutex);
	for (i = 0; i < request_queue_len; i++)
	{
		if (request_queue[i] == request) break;
	}
	if (i < request_queue_len)
	{
		request_queue_len--;
		for (i = i; i < request_queue_len; i++)
		{
			request_queue[i] = request_queue[i + 1];
		}
	}
	apr_thread_cond_signal(request_queue_cond);
	apr_thread_mutex_unlock(request_queue_mutex);
}

static void enqueue_sendreq(int source_world_rank, Envelope *envelope,
	int message_size)
{
	SendReq *sendreq = (SendReq *) apr_palloc(cm_pool, sizeof(SendReq));
	sendreq->source_world_rank = source_world_rank;
	msg_env_copy(&(sendreq->envelope), envelope);
	sendreq->message_size = message_size;
	apr_thread_mutex_lock(sendreq_queue_mutex);
	while (sendreq_queue_len == CM_MAX_REQUESTS)
	{
		apr_thread_cond_wait(sendreq_queue_cond, sendreq_queue_mutex);
	}
	sendreq_queue[sendreq_queue_len] = sendreq;
	sendreq_queue_len++;
	apr_thread_cond_signal(sendreq_queue_cond);
	apr_thread_mutex_unlock(sendreq_queue_mutex);
}

static void dequeue_sendreq(SendReq *sendreq)
{
	int i;
	apr_thread_mutex_lock(sendreq_queue_mutex);
	for (i = 0; i < sendreq_queue_len; i++)
	{
		if (sendreq_queue[i] == sendreq) break;
	}
	if (i < sendreq_queue_len)
	{
		sendreq_queue_len--;
		for (i = i; i < sendreq_queue_len; i++)
		{
			sendreq_queue[i] = sendreq_queue[i + 1];
		}
	}
	apr_thread_cond_signal(sendreq_queue_cond);
	apr_thread_mutex_unlock(sendreq_queue_mutex);
}

static Request *find_request(RequestType type, RequestStatus status,
	Envelope *request_env, Envelope *matched_env)
{
	int i, matched;
	Request *request = NULL;
	Envelope *sender, *receiver;
	
	apr_thread_mutex_lock(request_queue_mutex);
	
	matched = 0;
	for (i = 0; i < request_queue_len && !matched; i++)
	{
		request = request_queue[i];
		if (request->req_type != type) continue;
		if (request->status != status) continue;
		if (request_env != NULL)
		{
			if (request->req_type == SEND)
			{
				sender = &(request->request_env);
				receiver = request_env;
			}
			else
			{
				sender = request_env;
				receiver = &(request->request_env);
			}
			if (!msg_env_match(sender, receiver)) continue;
		}
		if (matched_env != NULL)
		{
			sender = matched_env;
			receiver = &(request->matched_env);
			if (!msg_env_match(sender, receiver)) continue;
		}
		matched = 1;
	}
	
	apr_thread_mutex_unlock(request_queue_mutex);
	
	return (matched) ? request : NULL;
}

static SendReq *find_sendreq(Envelope *request_env)
{
	int i, matched;	
	SendReq *sendreq = NULL;
	
	apr_thread_mutex_lock(sendreq_queue_mutex);
	
	matched = 0;
	for (i = 0; i < sendreq_queue_len && !matched; i++)
	{
		sendreq = sendreq_queue[i];
		if (msg_env_match(&(sendreq->envelope), request_env))
		{
			matched = 1;
			break;
		}
	}
	
	apr_thread_mutex_unlock(sendreq_queue_mutex);
	
	return (matched) ? sendreq : NULL;
}
